package com.ibeifeng.bigdata.senior.hadoop.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.ibeifeng.bigdata.senior.hadoop.mapreduce.WordCountMapReduce.WordCountReducer;

public class NoReduceMapReduce {

	// step 1: Mapper Class
	/**
	 * public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
	 */
	public static class WordCountMapper extends
			Mapper<LongWritable, Text, Text, IntWritable> {
		private Text mapOutputKey = new Text();
		private final static IntWritable mapOutputValue = new IntWritable(1);
		
		/**
		 * protected void map(KEYIN key, VALUEIN value, Context context)
		 */
		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
//			System.out.println("map-in-key:" + key.get() + " -- "//
//					+ "map-in-value:" + value.toString());
			
			
			// line value
			String lineValue = value.toString();
			
			// split
			String[] strs = lineValue.split(" ") ;
			
			// iterator
			for(String str: strs){
				// set map output key
				mapOutputKey.set(str);
				
				// output
				context.write(mapOutputKey, mapOutputValue);
				System.out.println("<" + mapOutputKey + "," +  mapOutputValue + ">");
			}
		}

	}

	// step 2: Reduce Class
		/**
		 * public class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT>
		 */
		public static class WordCountCombiner extends
				Reducer<Text, IntWritable, Text, IntWritable> {
			private IntWritable outputValue = new IntWritable();

			/**
			 *  protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context )
			 */
			@Override
			public void reduce(Text key, Iterable<IntWritable> values,
					Context context) throws IOException, InterruptedException {
				System.out.println("key = " + key);
				
				// temp : sum
				int sum = 0 ;
				
				// iterator
				for(IntWritable value : values){
					// total
					sum += value.get() ;
					
					System.out.print( value.get() + " " );
				}
				
				System.out.println();
				
				// set output value 
				outputValue.set(sum);
				
				// output
				context.write(key, outputValue);
			}

		}
	
	// step 3: Driver
	public int run(String[] args) throws Exception {
		// 1: get Configuration
		Configuration configuration = new Configuration();

		// 2: create job
		Job job = Job.getInstance(configuration,//
			this.getClass().getSimpleName()//
		);
		job.setJarByClass(this.getClass());
		
		// 3: set job
		// input  -> map  -> reduce -> output
		// 3.1: input
		Path inPath = new Path(args[0]);
		FileInputFormat.addInputPath(job, inPath);
		
		// 3.2: mapper
		job.setMapperClass(WordCountMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		// 3.3: reducer
		job.setNumReduceTasks(0);
		
		job.setCombinerClass(WordCountCombiner.class);
		
		
		// 3.4: output
		Path outPath = new Path(args[1]);
		FileOutputFormat.setOutputPath(job, outPath);
		
		// 4: submit job
		boolean isSuccess = job.waitForCompletion(true) ;
		
		return isSuccess ? 0 : 1 ;
	}
	
	
	public static void main(String[] args) throws Exception {
		/***/
		args = new String[]{ //
			"hdfs://bigdata-senior01.ibeifeng.com:8020/user/beifeng/input" ,//
			"hdfs://bigdata-senior01.ibeifeng.com:8020/user/beifeng/noreduceoutput222"
		};
		
		
		// run job
		int status = new NoReduceMapReduce().run(args) ;
		
		System.exit(status);
	}

}
